package com.sunfeng.two.p3;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import util.ConnectionUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * @Author : 清风冷影
 * @Description: 1、设置要求队列必须持久化
 * 2、设置要求消息必须持久化
 * 3、发布确认     消息直接发送给队列
 * 确认发布有三种模式
 * 单个发布确认： 发布速度特别慢
 * 批量发布确认
 * 异步发布确认
 * @Date : 2021/8/20  15:09
 */
public class ConfirmMessage {
    // 批量发消息的个数
    public static final int message_count = 1000;

    public static void main(String[] args) throws Exception {
        // 单个确认模式
        publishMessageIndividually();
        // 批量确认模式
        // publishMessageBatch();
        // 异步确认模式
        // publishMessageAsync();

    }

    public static void publishMessageIndividually() throws Exception {

        final Channel channel = ConnectionUtil.getChannel();
        // 开启发布确认
        channel.confirmSelect();

        String queueName = UUID.randomUUID().toString();
        //声明 队列持久化
        channel.queueDeclare(queueName, true, false, false, null);
        // 开始时间
        long start = System.currentTimeMillis();

        for (int i = 0; i < message_count; i++) {
            String message = i + "";
            //MessageProperties.PERSISTENT_TEXT_PLAIN; 设置消息的持久化
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            boolean waitForConfirms = channel.waitForConfirms();

            if (waitForConfirms) {
                System.out.println("消息发送成功");
            } else {
                System.out.println("消息发送失败");
            }
        }
        // 结束时间
        long end = System.currentTimeMillis();

        System.out.println("发送1000条单独确认的消息" + (end - start) + "ms");
    }

    public static void publishMessageBatch() throws Exception {

        final Channel channel = ConnectionUtil.getChannel();
        // 开启发布确认
        channel.confirmSelect();

        String queueName = UUID.randomUUID().toString();
        //声明 队列持久化
        channel.queueDeclare(queueName, true, false, false, null);
        // 开始时间
        long start = System.currentTimeMillis();

        for (int i = 0; i < message_count; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));

        }
        boolean waitForConfirms = channel.waitForConfirms();

        if (waitForConfirms) {
            System.out.println("消息发送成功");
        } else {
            System.out.println("消息发送失败");
        }
        // 结束时间
        long end = System.currentTimeMillis();

        System.out.println("发送1000条批量确认的消息" + (end - start) + "ms");
    }




    // 异步确认
    public static void publishMessageAsync() throws Exception {

        final Channel channel = ConnectionUtil.getChannel();
        // 开启异步确认模式
        channel.confirmSelect();


        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        String queueName = UUID.randomUUID().toString();
        //声明 队列持久化
        channel.queueDeclare(queueName, true, false, false, null);

        /**
         * 消息的标识
         *
         */
        ConfirmCallback confirmCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                //
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                //清除该部分未确认消息
                confirmed.clear();
            }else{
                //只清除当前序列号的消息
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("异步处理");
        };
        ConfirmCallback failCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("发布的消息"+message+"未被确认，序列号"+deliveryTag);
        };


        // 准备消息的监听器 监听那些消息成功了  哪些消息失败了
        channel.addConfirmListener(confirmCallback, failCallback);

        // 开始时间
        long start = System.currentTimeMillis();

        // 批量发送消息
        for (int i = 0; i < message_count; i++) {
            String message = i + "";
            // 将消息放回队列
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            // 此处需要记录发送的消息
        }

        // 结束时间
        long end = System.currentTimeMillis();

        System.out.println("发送1000条异步确认认的消息" + (end - start) + "ms");
    }
}
